package pulsar

import (
	"context"
	"github.com/apache/pulsar-client-go/pulsar"
	"github.com/apache/pulsar-client-go/pulsar/log"
	"github.com/pkg/errors"
	"go.opencensus.io/trace"
	"knowledge_admin/pkg/adapters/queue/pubsub"
	"net"
	"time"
)

type pub struct {
	client   pulsar.Client
	producer pulsar.Producer
	topic    string
	retry    int
}

func (p *pub) Close() {
	p.producer.Close()
	p.client.Close()
}

type SpanEnd func()

func (p *pub) Send(ctx context.Context, message pubsub.ProducerMessage) (err error) {
	pmsg, end := p.localMessageToPulsarMessage(ctx, message)
	defer end()
	return p.send(ctx, pmsg)
}

func (p *pub) send(ctx context.Context, msg pulsar.ProducerMessage) error {
	c := make(chan error, 1)

	p.producer.SendAsync(ctx, &msg, func(id pulsar.MessageID, message *pulsar.ProducerMessage, err error) { c <- err; close(c) })

	select {
	case <-ctx.Done():
		return ctx.Err()

	case cm := <-c:
		return cm
	}
}

func (p *pub) needRetry(err error) (need bool, duration time.Duration) {
	if err == nil {
		return
	}
	switch err.(type) {
	case *net.OpError:
		return true, 0
	case *pulsar.Error:
		perr := err.(*pulsar.Error)
		switch perr.Result() {
		case pulsar.ProducerQueueIsFull:
			duration = time.Millisecond * 100
			need = true
		case pulsar.TimeoutError, pulsar.ConnectError, pulsar.ReadError:
			need = true
		}
	}

	return
}

func (p *pub) SendAsync(
	ctx context.Context,
	msg pubsub.ProducerMessage,
	callback func(msg pubsub.ProducerMessage, err error),
) {
	pmsg, end := p.localMessageToPulsarMessage(ctx, msg)
	retry := p.retry

	var cb func(_ pulsar.MessageID, _ *pulsar.ProducerMessage, err error)
	cb = func(_ pulsar.MessageID, _ *pulsar.ProducerMessage, err error) {
		if retry > 0 {
			if need, duration := p.needRetry(err); need {
				retry--
				go func() {
					if duration > 0 {
						time.Sleep(duration)
					}
					p.producer.SendAsync(ctx, &pmsg, cb)
				}()
				return
			}
		}

		end()
		if callback != nil {
			callback(msg, err)
		}
	}

	p.producer.SendAsync(ctx, &pmsg, cb)
}

func (p *pub) localMessageToPulsarMessage(
	ctx context.Context,
	msg pubsub.ProducerMessage,
) (pulsar.ProducerMessage, SpanEnd) {
	pmsg := pulsar.ProducerMessage{
		Payload:             msg.Payload,
		Value:               msg.Value,
		EventTime:           msg.EventTime,
		Key:                 msg.Key,
		Properties:          msg.Properties,
		ReplicationClusters: msg.ReplicationClusters,
		SequenceID:          &msg.SequenceID,
	}

	if pmsg.Properties == nil {
		pmsg.Properties = make(map[string]string)
	}

	span := trace.FromContext(ctx)
	if span != nil {
		ctx, span = trace.StartSpan(ctx, p.topic, trace.WithSpanKind(trace.SpanKindClient))
		span.AddAttributes(trace.StringAttribute("topic", p.topic))
	}
	return pmsg, func() {
		if span != nil {
			span.End()
		}
	}
}

func NewPublisher(opt PubOptions) (pubsub.Publisher, error) {
	if opt.Logger == nil {
		opt.Logger = log.DefaultNopLogger()
	}
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		Logger: opt.Logger,
		URL:    opt.URL,
	})
	if err != nil {
		return nil, errors.New("failed to create pulsar client")
	}

	if opt.BatchingMaxPublishDelay == 0 {
		opt.BatchingMaxPublishDelay = 10
	}
	if opt.BatchingMaxMessages == 0 {
		opt.BatchingMaxMessages = 1000
	}

	producer, err := client.CreateProducer(pulsar.ProducerOptions{
		DisableBatching:         opt.DisableBatching,
		Topic:                   opt.Topic,
		Name:                    opt.Name,
		SendTimeout:             opt.SendTimeout * time.Second,
		MaxPendingMessages:      opt.MaxPendingMessages,
		DisableBlockIfQueueFull: !opt.BlockIfQueueFull,
		CompressionType:         pulsar.CompressionType(opt.CompressionType),
		BatcherBuilderType:      opt.BatcherBuilderType,
		BatchingMaxPublishDelay: time.Millisecond * time.Duration(opt.BatchingMaxPublishDelay),
		BatchingMaxMessages:     opt.BatchingMaxMessages,
		BatchingMaxSize:         1024 * opt.BatchingMaxMessages, //default  1M
	})
	if err != nil {
		return nil, errors.Wrap(err, "failed to create producer")
	}

	return &pub{
		client:   client,
		producer: producer,
		topic:    opt.Topic,
		retry:    opt.Retry,
	}, nil
}
